const { BackPressureTransform } = require('../assist');
const _ = require('lodash');
const moment = require('moment');
const utils = require('../../utils');

function sortObject(obj) {
  if (!_.isObject(obj)) {
    return obj;
  }
  const ret = {};
  const keys = _.keys(utils.omitNil(obj)).sort();
  keys.forEach(k => {
    ret[k] = _.isPlainObject(obj[k]) ? sortObject(obj[k]) : obj[k];
  });
  return ret;
}

module.exports = function (func) {
  const stream = new BackPressureTransform();
  func = utils.parseFunction(func);
  const uniq = {};

  stream._transform = async function (chunk, encoding, callback) {
    utils.debug('uniqueTransform [chunk]:', chunk);
    const md5 = utils.md5(JSON.stringify(sortObject(func ? func(chunk, { _, moment, utils }): chunk)));
    if (!uniq[md5]) {
      uniq[md5] = 1;
      stream.backPressurePush(chunk, callback);
    } else {
      callback();
    }
  };

  stream.on('finish', function () {
    utils.debug('uniqueTransform [finish]');
  });

  return stream;
};